/**
 * Exchanger类的使用
 *
 * @author fireway
 * @since 2019年 03月 14日 星期四 07:21:26 CST
 */
package concurrency;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import net.mindview.util.BasicGenerator;
import net.mindview.util.Generator;

class ExchangerProducer<T> implements Runnable {
    private Exchanger<List<T>> mExchanger;

    private Generator<T> mGenerator;

    private List<T> mHolder;

    private static final boolean DEBUG = true;

    ExchangerProducer(Exchanger<List<T>> exchg, Generator<T> gen, List<T> holder) {
        mExchanger = exchg;
        mGenerator = gen;
        mHolder = holder;
        if (DEBUG) {
            System.out.println("ExchangerProducer.mHolder[" + System.identityHashCode(mHolder) + "]");
        }
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                for (int i = 0; i < ExchangerDemo.sSize; i++) {
                    mHolder.add(mGenerator.next());
                }
                // Exchange full for empty:

                if (DEBUG) {
                    System.out.println("ExchangerProducer before exchange mHolder[" + System.identityHashCode(mHolder) + "]");
                    mHolder = mExchanger.exchange(mHolder);
                    System.out.println("ExchangerProducer after exchange mHolder[" + System.identityHashCode(mHolder) + "]");
                } else {
                    mHolder = mExchanger.exchange(mHolder);
                }
            }
        } catch (InterruptedException e) {
            // OK to terminate this way.
            // e.printStackTrace();
        }
    }
}

class ExchangerConsumer<T> implements Runnable {
    private Exchanger<List<T>> mExchanger;

    private List<T> mHolder;

    private volatile T mValue;

    private static final boolean DEBUG = true;

    ExchangerConsumer(Exchanger<List<T>> ex, List<T> holder) {
        mExchanger = ex;
        mHolder = holder;
        if (DEBUG) {
            System.out.println("ExchangerConsumer.mHolder[" + System.identityHashCode(mHolder) + "]");
        }
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                mHolder = mExchanger.exchange(mHolder);
                for(T x : mHolder) {
                    mValue = x;  // Fetch out value
                    mHolder.remove(x);  // OK for CopyOnWriteArrayList
                }
            }
        } catch (InterruptedException e) {
            // OK to terminate this way.
            // e.printStackTrace();
        }
        System.out.println("Final value: " + mValue);
    }
}

public class ExchangerDemo {
    /* package */static int sSize = 10;

    private static int sDelay = 5; // Seconds

    public static void main(String[] args) {
        if (args.length > 0) {
            sSize = new Integer(args[0]);
        }

        if (args.length > 1) {
            sDelay = new Integer(args[1]);
        }

        ExecutorService exec = Executors.newCachedThreadPool();
        Exchanger<List<Fat>> xc = new Exchanger<List<Fat>>();

        List<Fat> producerList = new CopyOnWriteArrayList<Fat>();
        List<Fat> consumerList = new CopyOnWriteArrayList<Fat>();

        Generator<Fat> gen = BasicGenerator.create(Fat.class);
        ExchangerProducer<Fat> exchangerProducer = new ExchangerProducer<Fat>(xc, gen, producerList);
        exec.execute(exchangerProducer);

        ExchangerConsumer<Fat> exchangerConsumer = new ExchangerConsumer<Fat>(xc, consumerList);
        exec.execute(exchangerConsumer);

        try {
            TimeUnit.SECONDS.sleep(sDelay);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        exec.shutdownNow();
    }
}
